package com.qch.kafka.demo;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.Collection;
import java.util.HashMap;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;

/**
 * @Author qch
 * @Date 2021/6/4
 */

public class MyAdminClient {
    private KafkaAdminClient client;
    @Before
    public void before(){
        HashMap<String, Object> configs = new HashMap<>();
        configs.put("bootstrap.servers","192.168.142.135:9092");
        configs.put("client.id","admin_001");
        client = (KafkaAdminClient)KafkaAdminClient.create(configs);
    }
    @After
    public void after(){
        //关闭客户端
        client.close();
    }
    @Test
    public void testListTopics() throws ExecutionException, InterruptedException {
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        listTopicsOptions.listInternal(true);
        listTopicsOptions.timeoutMs(500);
        ListTopicsResult listTopicsResult = client.listTopics(listTopicsOptions);
     /*
        Set<String> strings = listTopicsResult.names().get();
        strings.forEach(name->{
            System.out.println(name);
        });*/

        Collection<TopicListing> topicListings = listTopicsResult.listings().get();
        topicListings.forEach(new Consumer<TopicListing>() {
            @Override
            public void accept(TopicListing topicListing) {
                //该主题是否时内部主题
                boolean internal = topicListing.isInternal();
                String name = topicListing.name();
                System.out.println(topicListing);
            }
        });
    }
}
